package com.dcits.nifi;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.dbcp.DBCPService;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.*;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.StreamCallback;
import org.apache.nifi.processor.util.StandardValidators;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Tags({"cdc","oracle"})
@CapabilityDescription("")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute = "", description = "")})
@WritesAttributes(@WritesAttribute(attribute = "subscription", description = "the name of the subscription"))
public class CaptureChangeOracle extends AbstractProcessor {
    public static final PropertyDescriptor PROP_CDC_SOURCE_DB = new PropertyDescriptor
            .Builder().name("Source Database")
            .description("the source database to be captured")
            .required(true)
            .identifiesControllerService(DBCPService.class)
            .build();

    public static final PropertyDescriptor PROP_SUBSCRIPTION = new PropertyDescriptor
            .Builder().name("Subscription Name")
            .description("the name of the subscription")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor PROP_SOURCE_SQL = new PropertyDescriptor
            .Builder().name("Query Sql")
            .description("the sql used for query data from the subscriber_view")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor MAXRECORDS_PER_FLOWFILE = new PropertyDescriptor
            .Builder().name("Flowfile Size")
            .description("the maximum number of records per flowfile")
            .required(true)
            .defaultValue("1000")
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .build();

    public static final Relationship REL_SUCCESS = new Relationship.Builder()
            .name("success")
            .description("Success, all done")
            .build();

    private List<PropertyDescriptor> descriptors;

    private Set<Relationship> relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
        descriptors.add(PROP_CDC_SOURCE_DB);
        descriptors.add(PROP_SUBSCRIPTION);
        descriptors.add(PROP_SOURCE_SQL);
        descriptors.add(MAXRECORDS_PER_FLOWFILE);
        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set<Relationship> relationships = new HashSet<Relationship>();
        relationships.add(REL_SUCCESS);
        this.relationships = Collections.unmodifiableSet(relationships);
    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {

    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        List<FlowFile> flowFiles = new ArrayList<>();

        final String subName = context.getProperty(PROP_SUBSCRIPTION).getValue();  //订阅者名称
        final int maxNum = context.getProperty(MAXRECORDS_PER_FLOWFILE).asInteger();  //每个flowfile的最大记录条数
        final String querySql = context.getProperty(PROP_SOURCE_SQL).getValue();  //待查询的sql
        final DBCPService cdcSourceDbcpService = context.getProperty(PROP_CDC_SOURCE_DB).asControllerService(DBCPService.class);

        try(Connection con = cdcSourceDbcpService.getConnection(); Statement state = con.createStatement();){
            state.execute("DBMS_CDC_SUBSCRIBE.EXTEND_WINDOW( subscription_name => '"+subName+"')");
            ResultSet res = state.executeQuery(querySql);
            ResultSetMetaData resMeta = res.getMetaData();
            JSONArray contentArray = new JSONArray();
            while(res.next()) {
                JSONObject object = new JSONObject();
                for(int i=0;i<resMeta.getColumnCount();i++){
                    String colName = resMeta.getColumnName(i+1);
                    Object value = res.getObject(i+1);
                    value = value==null?"null":value;
                    object.put(colName,value);
                }
                contentArray.add(object);
                if(contentArray. size() == maxNum){
                    FlowFile flowFile = newFlowFile(session);
                    session.putAttribute(flowFile,"subscription",subName);
                    session.write(flowFile, new StreamCallback(){
                        public void process(InputStream in, OutputStream out) throws IOException {
                            out.write(contentArray.toJSONString().getBytes(Charset.forName("UTF-8")));
                        }
                    });
                    flowFiles.add(flowFile);
                    contentArray.clear();
                }
            }
            if(contentArray. size() != 0){
                FlowFile flowFile = newFlowFile(session);
                session.putAttribute(flowFile,"subscription",subName);
                session.write(flowFile, new StreamCallback(){
                    public void process(InputStream in, OutputStream out) throws IOException{
                        out.write(contentArray.toJSONString().getBytes(Charset.forName("UTF-8")));
                    }
                });
                flowFiles.add(flowFile);
                contentArray.clear();
            }
            session.transfer(flowFiles,REL_SUCCESS);
            res.close();
            state.execute("DBMS_CDC_SUBSCRIBE.PURGE_WINDOW( subscription_name => '"+subName+"')");
        }catch(Exception e){
            session.remove(flowFiles);
            getLogger().error("订阅["+subName+"]异常!",e);
        }
    }

    //确保能创建新的flowFile
    private  FlowFile newFlowFile(ProcessSession session){
        FlowFile flowFile = null;
        while(true){
            try {
                flowFile = session.create();
                if (flowFile != null) {
                    break;
                }else{
                    TimeUnit.SECONDS.sleep(5);
                }
            }catch (Exception e){
                getLogger().info("创建flowFile异常",e);
            }
        }
        return flowFile;
    }
}
